package com.hhf.rrd.wordcount;

import com.alibaba.fastjson.JSON;
import com.hhf.rrd.data.WordCountData;
import com.hhf.rrd.model.WordEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;

/**
 * WordCount 批次任务
 *
 * @author huanghaifeng15
 * @date 2022/2/9 20:30
 **/
@Slf4j
public class WordCountBatchApp {

    public static void main(String[] args) throws Exception {
        // 获取 运行上下文
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 获取 source <固定文件>
        DataSource<String> source1 = env.readTextFile("data/wc/wc.data");

        // 获取 source <列表>
        DataSource<String> source = env.fromCollection(Arrays.asList(WordCountData.WORDS2));

        // 执行 Transformation
        FlatMapOperator<String, String> flatMapOperator = source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                if (StringUtils.isNotBlank(value)){
                    String[] wordArr = value.split(",");
                    for (String word : wordArr) {
                        out.collect(word.trim());
                    }
                }
            }
        });

        MapOperator<String, Tuple2<String, Integer>> mapOperator = flatMapOperator.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                Tuple2<String, Integer> tuple2 = new Tuple2<>();
                tuple2.setFields(value,1);
                return tuple2;
            }
        });

        UnsortedGrouping<Tuple2<String, Integer>> unsortedGrouping = mapOperator.groupBy(0);

        AggregateOperator<Tuple2<String, Integer>> aggregateOperator = unsortedGrouping.sum(1);

        // 执行 sink
        List<Tuple2<String, Integer>> collect = aggregateOperator.collect();
        for (Tuple2<String, Integer> tuple2 : collect) {
            WordEntity wordEntity = new WordEntity();
            wordEntity.setWord(tuple2.f0);
            wordEntity.setFrequency(tuple2.f1);
            log.info(JSON.toJSONString(wordEntity,true));
        }

        aggregateOperator.print();
    }
}
